package databus.client.consumer.asy.group;

import com.google.protobuf.InvalidProtocolBufferException;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import studio.raptor.databus.client.DatabusConfig;
import studio.raptor.databus.client.DatabusConsumer;
import studio.raptor.databus.client.DatabusConsumerBuilder;
import studio.raptor.databus.client.EventMessage;
import studio.raptor.databus.command.Event;

import javax.jms.JMSException;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Objects;
import java.util.concurrent.locks.LockSupport;

/**
 * 应用内组消费者消费消息-异步消费
 *
 * @author koalton
 *
 */
public class CustomerCenterDLQ {

    DatabusConsumerBuilder consumerBuilder = null ;
    public static final SimpleDateFormat Y2MS = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    Logger log = LoggerFactory.getLogger(CustomerCenterDLQ.class);
    //两次获取消息的最大时间间隔-60s
    public static final int MAX_WAIT_RECEIVE_GAP_TIME = 60;
    //两次获取消息的初始化时间间隔-1s
    public static final int INIT_RECEIVE_GAP_TIME = 1;
    //两次获取消息时间间隔增长步长 2s
    public static final int RECEIVE_GAP_TIME_STEP = 2;

    public CustomerCenterDLQ() {
        try {
            //实例配置对象
            DatabusConfig config = new DatabusConfig("Customer");
//            DatabusConfig config = new DatabusConfig("so_order");
            //通过配置实例化builder对象
            consumerBuilder = new DatabusConsumerBuilder(config);
        } catch (JMSException e1) {
            log.info(e1.toString());
            e1.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
            log.info(e.getMessage());
            e.printStackTrace();
        }
    }

    /**
     * 从服务端获取消息
     */
    @Test
    public void receive() {
        DatabusConsumer consumer = null;
        try {
            //创建消费事件[100001-AddOrder]的消费者,此消费者自动ACK消息
            String groupName = "";
            groupName = "group2";
            groupName = "group1";
            groupName = "group3";
            consumer =
                    this.initConsumer(
                            consumerBuilder,
                            "AddOrder",
                            100001,
                            null,
                            groupName);
        } catch (JMSException e) {
            e.printStackTrace();
        }
        if( null == consumer ){
            System.out.println("消费者初始化失败");
            return ;
        }
        try {
            int waitTime = INIT_RECEIVE_GAP_TIME;
            while (true) {
                EventMessage eventMessage = consumer.receiveUnACKMessage(1*1000);
                if (eventMessage == null) {
                    System.out.println("At " + Y2MS.format(new Date()) + " receive:The Queue is empty");
                    LockSupport.parkUntil(System.currentTimeMillis() + waitTime * 1000);
                    waitTime += RECEIVE_GAP_TIME_STEP;
                    if( waitTime >  MAX_WAIT_RECEIVE_GAP_TIME ){
                        waitTime = MAX_WAIT_RECEIVE_GAP_TIME;
                    }
                    continue;
                }
                //eventMessage.acknowledge();
                consumer.notifyRedeliver();
                waitTime = 1 ;
                System.out.println("At " + Y2MS.format(new Date()) + " receive:" + eventCoreInfo(eventMessage.getEvent()));
            }
        } catch (JMSException e) {
            log.info(e.getMessage());
            e.printStackTrace();
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
        LockSupport.parkUntil(System.currentTimeMillis() + 30 * 60 * 1000);
    }

    /**
     * 从服务端获取消息时携带消息过滤器
     */
    @Test
    public void receiveWithSelector() {
        DatabusConsumer consumer = null;
        try {
            String selector= "DATABUS_MSG_PROP_BUSINESSNO='11704753639'";
            //创建消费事件[100001-AddOrder]的消费者,此消费者自动ACK消息
            String groupName = "";
            groupName = "group3";
            groupName = "group2";
            groupName = "group1";
            consumer =
                    this.initConsumer(
                            consumerBuilder,
                            "AddOrder",
                            100001,
                            selector,
                            groupName);
        } catch (JMSException e) {
            e.printStackTrace();
        }
        if( null == consumer ){
            System.out.println("消费者初始化失败");
            return ;
        }
        try {
            int waitTime = INIT_RECEIVE_GAP_TIME;
            while (true) {
                EventMessage eventMessage = consumer.receiveUnACKMessage(1*1000);
                if (eventMessage == null) {
                    System.out.println("At " + Y2MS.format(new Date()) + " receive:The Queue is empty");
                    LockSupport.parkUntil(System.currentTimeMillis() + waitTime * 1000);
                    waitTime += RECEIVE_GAP_TIME_STEP;
                    if( waitTime >  MAX_WAIT_RECEIVE_GAP_TIME ){
                        waitTime = MAX_WAIT_RECEIVE_GAP_TIME;
                    }
                    continue;
                }
                consumer.notifyRedeliver();
                //eventMessage.acknowledge();
                waitTime = 1 ;
                System.out.println("At " + Y2MS.format(new Date()) + " receive:" + eventCoreInfo(eventMessage.getEvent()));
            }
        } catch (JMSException e) {
            log.info(e.getMessage());
            e.printStackTrace();
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
        LockSupport.parkUntil(System.currentTimeMillis() + 30 * 60 * 1000);
    }

    public String eventCoreInfo(Event event) {
        //事件来源
        String eventSource = event.getEventBasicData().getSource();
        //事件序号
        String eventSeq = event.getEventBasicData().getKindId()+"";
        //事件名称
        String eventName = event.getEventBasicData().getName();
        //业务编号
        String businessNo = event.getEventMetaData().getBusinessNo()+"";
        //业务数据
        String centent = event.getEventBusinessData().getContent();
        return "{EventSource:" + eventSource + ",eventName:" + eventName + ",businessNo:" + businessNo + ",content :"+centent+"}";
    }

    /**
     * 实例消费者
     * @param consumerBuilder 消费者创建者
     * @param eventName 消费者所需消费的事件名称
     * @param eventKindId 消费者所需消费的事件kindId
     * @param consumerSelector 消费者所携带的消息过滤器，如果为NULL表示不会过滤消息
     * @param groupName 应用内组名称
     * @return
     */
    private DatabusConsumer initConsumer(
            DatabusConsumerBuilder consumerBuilder,
            String eventName,
            Integer eventKindId,
            String consumerSelector,
            String groupName) throws JMSException {
        if(Objects.equals(null,consumerBuilder)){
            return null ;
        }
        DatabusConsumer consumer =
                consumerBuilder.createDLQGroupConsumer(
                        eventName,eventKindId,groupName,consumerSelector);
        return consumer ;
    }
}
